Safe async cancellations. #880

tomchristie · 2024-02-06T11:34:36Z

Closes #830
Closes #785
Closes #861
Closes encode/httpx#1171

Let's talk about what's going on here.

We've got an issue with handling async cancellations correctly, which needs some re-working in order to comprehensively resolve it. We're somewhat in contrast to either urllib3 (sync) or aiohttp (async) here, because we're having to get both the thread-safe and the task-safe-plus-also-support-cancellations cases correct.

Currently we're at thread-safe plus task-safe, but missing correct handling of also allowing external cancellations at any point.

So then...

This pull request reworks the handling of the connection pool state. The state is a list of connections, a list of requests, and an association between each request and the connection that has been selected to handle it.

The fundamental change in this pull request is that the management of the pool state has been re-worked so that it does not include I/O.

httpcore/httpcore/_async/connection_pool.py

Lines 228 to 299 in cda040d

    
               def _assign_requests_to_connections(self) -> List[AsyncConnectionInterface]: 
        
                   """ 
        
                   Manage the state of the connection pool, assigning incoming 
        
                   requests to connections as available. 
        
                   Called whenever a new request is added or removed from the pool. 
        
                   Any closing connections are returned, allowing the I/O for closing 
        
                   those connections to be handled seperately. 
        
                   """ 
        
                   closing_connections = [] 
        
                   # First we handle cleaning up any connections that are closed, 
        
                   # have expired their keep-alive, or surplus idle connections. 
        
                   for connection in list(self._connections): 
        
                       if connection.is_closed(): 
        
                           # log: "removing closed connection" 
        
                           self._connections.remove(connection) 
        
                       elif connection.has_expired(): 
        
                           # log: "closing expired connection" 
        
                           self._connections.remove(connection) 
        
                           closing_connections.append(connection) 
        
                       elif ( 
        
                           connection.is_idle() 
        
                           and len([connection.is_idle() for connection in self._connections]) 
        
                           > self._max_keepalive_connections 
        
                       ): 
        
                           # log: "closing idle connection" 
        
                           self._connections.remove(connection) 
        
                           closing_connections.append(connection) 
        
                   # Assign queued requests to connections. 
        
                   queued_requests = [ 
        
                       request for request in self._requests if request.connection is None 
        
                   ] 
        
                   for pool_request in queued_requests: 
        
                       origin = pool_request.request.url.origin 
        
                       avilable_connections = [ 
        
                           connection 
        
                           for connection in self._connections 
        
                           if connection.can_handle_request(origin) and connection.is_available() 
        
                       ] 
        
                       idle_connections = [ 
        
                           connection for connection in self._connections if connection.is_idle() 
        
                       ] 
        
                       # There are three cases for how we may be able to handle the request: 
        
                       # 
        
                       # 1. There is an existing connection that can handle the request. 
        
                       # 2. We can create a new connection to handle the request. 
        
                       # 3. We can close an idle connection and then create a new connection 
        
                       #    to handle the request. 
        
                       if avilable_connections: 
        
                           # log: "reusing existing connection" 
        
                           connection = avilable_connections[0] 
        
                           pool_request.assign_to_connection(connection) 
        
                       elif len(self._connections) < self._max_connections: 
        
                           # log: "creating new connection" 
        
                           connection = self.create_connection(origin) 
        
                           self._connections.append(connection) 
        
                           pool_request.assign_to_connection(connection) 
        
                       elif idle_connections: 
        
                           # log: "closing idle connection" 
        
                           connection = idle_connections[0] 
        
                           self._connections.remove(connection) 
        
                           closing_connections.append(connection) 
        
                           # log: "creating new connection" 
        
                           connection = self.create_connection(origin) 
        
                           self._connections.append(connection) 
        
                           pool_request.assign_to_connection(connection) 
        
                   return closing_connections

This ensures that updating the state of the connection pool is always atomic.

As part of resolving this, also adds connection pool __repr__ to demonstate the state more clearly.

The look like so...

<httpcore.ConnectionPool [Requests: 10 active, 3 queued | Connections: 2 active, idle]>

tomchristie · 2024-02-07T14:54:37Z

tests/_async/test_connection_pool.py

            assert info == [
-                "<AsyncHTTPConnection ['http://example.com:80', HTTP/1.1, ACTIVE, Request Count: 1]>",
                "<AsyncHTTPConnection ['https://example.com:443', HTTP/1.1, IDLE, Request Count: 2]>",
+                "<AsyncHTTPConnection ['http://example.com:80', HTTP/1.1, ACTIVE, Request Count: 1]>",
            ]


The ordering of connections changes slightly here.
We're now working with a policy of... connections remain in the order they were created.

tomchristie · 2024-02-07T14:55:12Z

tests/_async/test_connection_pool.py

        assert info == [
-            "<AsyncHTTPConnection ['http://example.com:80', HTTP/1.1, IDLE, Request Count: 1]>",
            "<AsyncHTTPConnection ['https://example.com:443', HTTP/1.1, IDLE, Request Count: 2]>",
+            "<AsyncHTTPConnection ['http://example.com:80', HTTP/1.1, IDLE, Request Count: 1]>",
        ]


Again the ordering is slightly different from previous behaviour.
I would suggest this is more obvious.

tomchristie · 2024-02-07T14:55:54Z

tests/_async/test_connection_pool.py

        assert info == [
            "<AsyncHTTPConnection ['https://example.com:443', HTTP/2, IDLE, Request Count: 1]>",
-            "<AsyncHTTPConnection ['https://example.com:443', HTTP/2, CLOSED, Request Count: 1]>",
        ]


There is a marginal behavior change here.
We've (correctly) removed the closed connection from the pool.

tomchristie · 2024-02-07T15:01:18Z

httpcore/_async/connection_pool.py

            extensions=response.extensions,
        )

-    async def response_closed(self, status: RequestStatus) -> None:
+    def _assign_requests_to_connections(self) -> List[AsyncConnectionInterface]:


This is where we mange the state of the connection pool, entirely within a non-I/O block. The async case cannot have cancellations or context-switches midway through the state management. The sync case is explicitly guarded with a thread lock.

Review is probably best made against the module as a whole, rather than the diff.

https://github.com/encode/httpcore/blob/f91789652485d5ec7651458bc20a9164ff56dbd6/httpcore/_async/connection_pool.py

rattrayalex · 2024-02-09T19:01:40Z

FYI, a user reports that this branch does indeed fix issues they saw in the OpenAI Python client library: openai/openai-python#1059 (comment)

Thank you for the work on this so far!

tomchristie · 2024-02-09T21:27:51Z

Fantastic, thanks for the feedback.

karpetrosyan · 2024-02-10T17:41:27Z

Amazing work @tomchristie, this PR rsolves many problems..

karpetrosyan

I guess we also need a changelog here

tomchristie added 11 commits February 6, 2024 12:33

Connection pool work

8069f82

Merge branch 'master' into clean-state-cancellations

ab7a9aa

Connection pool work

6e3453e

Connection pool work

3791d5e

Connection pool work

8d9371b

Comments

fdd5e78

Comments

fcb0ca2

Connection pool work

cda040d

Reraise

d246c91

Lookin sharp

20f088d

nocover directive

f917896

tomchristie commented Feb 7, 2024

View reviewed changes

tomchristie marked this pull request as ready for review February 7, 2024 15:01

tomchristie requested a review from a team February 7, 2024 15:14

tomchristie changed the title ~~Connection pool work~~ Safe async cancellations. Feb 7, 2024

tomchristie added 2 commits February 8, 2024 16:04

Merge branch 'master' into clean-state-cancellations

d6f19ca

Safe cancellations

837f1e0

dumbPy mentioned this pull request Feb 9, 2024

Reusing AsyncOpenAI client results in openai.APIConnectionError openai/openai-python#1059

Closed

1 task

karpetrosyan approved these changes Feb 10, 2024

View reviewed changes

Update CHANGELOG

a42a30d

tomchristie mentioned this pull request Feb 12, 2024

PoolTimeout when num tasks in asyncio.gather() exceeds client max_connections encode/httpx#1171

Closed

tomchristie merged commit 7b04cda into master Feb 12, 2024
5 checks passed

tomchristie deleted the clean-state-cancellations branch February 12, 2024 11:20

tomchristie mentioned this pull request Feb 12, 2024

Version 1.0.3 #883

Merged

syeopite mentioned this pull request Feb 12, 2024

PoolTimeout exceptions on requests to Tumblr syeopite/priviblur#4

Closed

CzBiX mentioned this pull request Jun 12, 2024

Raise exception with cause exception #915

Open

3 tasks

amurais mentioned this pull request Oct 31, 2024

Request errors on overloaded systems may poison the connection pool #550

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Safe async cancellations. #880

Safe async cancellations. #880

tomchristie commented Feb 6, 2024 •

edited

Loading

tomchristie Feb 7, 2024

tomchristie Feb 7, 2024

tomchristie Feb 7, 2024

tomchristie Feb 7, 2024

rattrayalex commented Feb 9, 2024 •

edited

Loading

tomchristie commented Feb 9, 2024

karpetrosyan commented Feb 10, 2024

karpetrosyan left a comment

	def _assign_requests_to_connections(self) -> List[AsyncConnectionInterface]:
	"""
	Manage the state of the connection pool, assigning incoming
	requests to connections as available.

	Called whenever a new request is added or removed from the pool.

	Any closing connections are returned, allowing the I/O for closing
	those connections to be handled seperately.
	"""
	closing_connections = []

	# First we handle cleaning up any connections that are closed,
	# have expired their keep-alive, or surplus idle connections.
	for connection in list(self._connections):
	if connection.is_closed():
	# log: "removing closed connection"
	self._connections.remove(connection)
	elif connection.has_expired():
	# log: "closing expired connection"
	self._connections.remove(connection)
	closing_connections.append(connection)
	elif (
	connection.is_idle()
	and len([connection.is_idle() for connection in self._connections])
	> self._max_keepalive_connections
	):
	# log: "closing idle connection"
	self._connections.remove(connection)
	closing_connections.append(connection)

	# Assign queued requests to connections.
	queued_requests = [
	request for request in self._requests if request.connection is None
	]
	for pool_request in queued_requests:
	origin = pool_request.request.url.origin
	avilable_connections = [
	connection
	for connection in self._connections
	if connection.can_handle_request(origin) and connection.is_available()
	]
	idle_connections = [
	connection for connection in self._connections if connection.is_idle()
	]

	# There are three cases for how we may be able to handle the request:
	#
	# 1. There is an existing connection that can handle the request.
	# 2. We can create a new connection to handle the request.
	# 3. We can close an idle connection and then create a new connection
	# to handle the request.
	if avilable_connections:
	# log: "reusing existing connection"
	connection = avilable_connections[0]
	pool_request.assign_to_connection(connection)
	elif len(self._connections) < self._max_connections:
	# log: "creating new connection"
	connection = self.create_connection(origin)
	self._connections.append(connection)
	pool_request.assign_to_connection(connection)
	elif idle_connections:
	# log: "closing idle connection"
	connection = idle_connections[0]
	self._connections.remove(connection)
	closing_connections.append(connection)
	# log: "creating new connection"
	connection = self.create_connection(origin)
	self._connections.append(connection)
	pool_request.assign_to_connection(connection)

	return closing_connections

Safe async cancellations. #880

Safe async cancellations. #880

Conversation

tomchristie commented Feb 6, 2024 • edited Loading

tomchristie Feb 7, 2024

Choose a reason for hiding this comment

tomchristie Feb 7, 2024

Choose a reason for hiding this comment

tomchristie Feb 7, 2024

Choose a reason for hiding this comment

tomchristie Feb 7, 2024

Choose a reason for hiding this comment

rattrayalex commented Feb 9, 2024 • edited Loading

tomchristie commented Feb 9, 2024

karpetrosyan commented Feb 10, 2024

karpetrosyan left a comment

Choose a reason for hiding this comment

tomchristie commented Feb 6, 2024 •

edited

Loading

rattrayalex commented Feb 9, 2024 •

edited

Loading